package com.event.ingestion.kafka;

import com.event.ingestion.common.HBaseWriter;
import com.event.ingestion.common.MongoWriter;
import com.event.ingestion.common.Parsable;
import com.event.ingestion.common.Persistable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @author ZzHh
 * @Classname UserConsumer
 * @Description TODO
 * @Date:2020/06/18 12:29
 * @Create by IntelliJ IDEA
 **/

/**
 * userConsumer业务代码编写
 */
public class UserConsumer extends BaseConsumer {
//    //hbae parser
//    public static class UserHBaseParse implements Parsable<Put> {
//        @Override
//        public Boolean isHeader(String[] fields) {
//            return (isValid(fields) && fields[0].equals("user_id") && fields[1].equals("locale") && fields[2].equals("birthyear") && fields[3].equals("gender") && fields[4].equals("joinedAt") && fields[5].equals("location") && fields[6].equals("timezone"));
//        }
//
//        @Override
//        public Boolean isValid(String[] fields) {
//            return (fields.length > 6);
//        }
//
//        //3197468391,id_ID,1993,male,2012-10-02T06:40:55.524Z,Medan  Indonesia,480
//        //解析数据,封装为Put
//        @Override
//        public Put parse(String[] fields) {
//            Put p = new Put(Bytes.toBytes(fields[0]));  //rowkey
//
//            //profile
//            p = p.addColumn(Bytes.toBytes("profile"),Bytes.toBytes("birth_year"),Bytes.toBytes(fields[2]));
//            p = p.addColumn(Bytes.toBytes("profile"),Bytes.toBytes("gender"),Bytes.toBytes(fields[3]));
//
//            //region
//            p = p.addColumn(Bytes.toBytes("region"),Bytes.toBytes("locale"),Bytes.toBytes(fields[1]));
//            p = p.addColumn(Bytes.toBytes("region"),Bytes.toBytes("location"),Bytes.toBytes(fields[5]));
//            p = p.addColumn(Bytes.toBytes("region"),Bytes.toBytes("time_zone"),Bytes.toBytes(fields[6]));
//
//            //registration:joined_at
//            p = p.addColumn(Bytes.toBytes("registration"),Bytes.toBytes("joined_at"),Bytes.toBytes(fields[4]));
//
//            return p;
//        }
//    }

    public UserConsumer() {
        //Persistable  HBaseTable
        //super(new Persistable [] {new HBaseWriter("events_db_users",new UserHBaseParse())});
        super(new Persistable [] {new HBaseWriter("events_db:users",new com.event.ingestion.data.hbase.UserParser()),
        new MongoWriter("users",new com.event.ingestion.data.mongo.UserParser())});
    }

    @Override
    protected String getKafkaConsumerGrp() {
        return "grpUsers";
    }

    @Override
    protected String getKafkaTopic() {
        return "users";
    }

    @Override
    protected Boolean getKafkaAutoCommit() {
        return false;
    }
}